package com.xujianlong.day09;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Flink01_SQL_GroupWindow {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("create table sensor(" +
                "id string,"+
                "ts bigint,"+
                "vc int,"+
                "t as to_timestamp(from_unixtime(ts/1000,'yyyy-MM-dd HH:mm:ss')),"+
                "watermark for t as t - interval '5' second)"+
                "with("
                +"'connector'='filesystem',"
                +"'path'='input/sensor-sql.txt',"
                +"'format'='csv'"
                +")"
                );

//        tableEnv.executeSql("select id," +
//                "tumble_start(t,interval '2' second) as wStart, tumble_end(t,interval '2' second) as wEnd, " +
//                "sum(vc) sum_vc " +
//                "from sensor " +
//                "group by tumble(t,interval '2' second),id")
//                .print();
//        tableEnv.executeSql("select id," +
//                "hop_start(t,interval '2' second,interval '3' second) as wStart, hop_end(t,interval '2' second,interval '3' second) as wEnd, " +
//                "sum(vc) sum_vc " +
//                "from sensor " +
//                "group by hop(t,interval '2' second,interval '3' second),id")
//                .print();

        tableEnv.executeSql("select id," +
                "session_start(t,interval '2' second) as wStart, session_end(t,interval '2' second) as wEnd, " +
                "sum(vc) sum_vc " +
                "from sensor " +
                "group by session(t,interval '2' second),id")
                .print();
    }
}
